package com.yuyou.fn.mq;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.yuyou.fn.common.rocketmq.Consumer;
import com.yuyou.fn.common.service.ISmsService;
import com.yuyou.fn.common.util.BeanUtils;
import com.yuyou.fn.common.util.JSONS;
import com.yuyou.fn.common.util.Utils;
import com.yuyou.fn.controller.AdminEnrollApiController;
import com.yuyou.fn.educational.entity.AttendClassRecord;
import com.yuyou.fn.educational.entity.Student;
import com.yuyou.fn.educational.entity.StudentReg;
import com.yuyou.fn.educational.service.*;
import com.yuyou.fn.electronicinvoice.service.IInvoiceService;
import com.yuyou.fn.platform.entity.CustomerUser;
import com.yuyou.fn.platform.entity.Oauth;
import com.yuyou.fn.platform.service.IOauthService;
import com.yuyou.fn.portal.entity.OuterOrder;
import com.yuyou.fn.portal.service.IOuterOrderService;
import com.yuyou.fn.portal.service.IPushMessageService;
import com.yuyou.fn.portal.vo.ClassInfoVo;
import com.yuyou.fn.settlement.entity.PosPayRecord;
import com.yuyou.fn.settlement.service.IAccountRechargeService;
import com.yuyou.fn.settlement.service.IPayRecordService;
import com.yuyou.fn.tradingservice.service.IServiceOrderItemService;
import com.yuyou.fn.tradingservice.service.IServiceOrderService;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.context.request.async.DeferredResult;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.rmi.RemoteException;
import java.util.*;

public class ConsumerHelper {

    private static final Logger log = LoggerFactory.getLogger(ConsumerHelper.class);

    @Resource
    private Consumer consumer ;

    @Resource
    private IInvoiceService invoiceService;


    @Resource
    private IClassLectureService classLectureService;

    @Resource
    private ICourseConsumeService courseConsumeService;


    @Resource
    private IOuterOrderService outerOrderService;

    @Resource
    private IPushMessageService pushMessageService;

    @Resource
    private ISmsService smsService;

    @Resource
    private IOauthService oauthService;

    @Resource
    private ICustomerUserService customerUserService;

    @Resource
    private IServiceOrderItemService serviceOrderItemService;

    @Resource
    private IServiceOrderService serviceOrderService;

    @Resource
    private IStudentRegService studentRegService ;

    @Resource
    private IPayRecordService payRecordService ;

    @Resource
    private IContinueClassStatService continueClassStatService;

    @Resource
    private IStudentService studentService ;

    @Resource
    private IAccountRechargeService accountRechargeService;

    @Resource
    private ICourseClassService courseClassService ;

    public void initSubscribe(){

        consumer.subscribe("order_pay_success", "serviceOrder","ServiceOrder", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt : list){
                    try{
                        JSONObject jsonObject = JSONS.toJSONObject(messageExt.getBody());
                        Long orderId  = jsonObject.getLong("busId");
                        BigDecimal paidAmount = jsonObject.getBigDecimal("amount");
                        serviceOrderService.payServiceOrder(orderId,paidAmount,1);
                    }catch (Exception e){
                        log.error("消息处理失败,topic=order_pay_success,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER :ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        /**
        *@Description: 线上余额充值
        *@Param: 
        *@return: 
        *@Author: lzh
        *@date: 
        */
        consumer.subscribe("order_pay_success", "mobileRechargeOrder","mobileRechargeOrder", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt : list){
                    try{
                        JSONObject jsonObject = JSONS.toJSONObject(messageExt.getBody());
                        Long orderId  = jsonObject.getLong("busId");
                        BigDecimal paidAmount = jsonObject.getBigDecimal("amount");
                        JSONObject payInfoExt = new JSONObject();
                        payInfoExt.put("transactionNo" ,jsonObject.getString("transactionNo"));
                        payInfoExt.put("channel" ,jsonObject.getString("channel"));
                        accountRechargeService.finishAccountRecharge(orderId,4,payInfoExt.toJSONString() ,null ,paidAmount);
                    }catch (Exception e){
                        log.error("消息处理失败,topic=order_pay_success,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER :ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });


        consumer.subscribe("order_pay_success", "serviceItemOrder","ServiceOrderItem", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt : list){
                    try{
                        JSONObject jsonObject = JSONS.toJSONObject(messageExt.getBody());
                        Long orderItemId = jsonObject.getLong("busId");
                        BigDecimal paidAmount = jsonObject.getBigDecimal("amount");
                        serviceOrderItemService.payServiceOrderItem(orderItemId,paidAmount,2);
                    }catch (Exception e){
                        log.error("消息处理失败,topic=order_pay_success,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER :ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.subscribe("order_success_pay_notify" , "PushMessageService", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt : list){
                    try{
                        JSONObject jsonObject = JSONS.toJSONObject(messageExt.getBody());
                        Long userId  = jsonObject.getLong("userId");
                        Long outerOrderId  = jsonObject.getLong("outerOrderId");
                        BigDecimal paidAmount = jsonObject.getBigDecimal("paidAmount");
                        OuterOrder outerOrder=outerOrderService.findById(outerOrderId);
                        if (outerOrder==null) continue;
                        Oauth oauth = oauthService.findByUserId(userId);
                        if (oauth!=null){
                            try{
                                pushMessageService.pushPaysSuccessMessageToWX(outerOrderId, paidAmount, outerOrder, oauth);
                            }catch (Exception e){
                                //ignore error
                            }

                        }


                        CustomerUser customerUser = customerUserService.findById(userId);
                        List<ClassInfoVo> classInfoVos = JSONObject.parseObject( outerOrder.getOrderData() , new TypeReference<List<ClassInfoVo>>(){});
                        Set<String> studentNames = new HashSet<>(classInfoVos.size());
                        List<String> classNames = new ArrayList<>(classInfoVos.size());
                        for (ClassInfoVo classInfoVo : classInfoVos ){
                            classNames.add(classInfoVo.getCourseClass().getClassName());
                            for(Student student : classInfoVo.getStudentList()){
                                studentNames.add(student.getName());
                            }
                        }

                        String content = String.format("恭喜您！学员%s已成功报名“%s”课程，请关注公众号“誉优品学”，并可查看课程详情。如有疑问，请联系报名校区，谢谢您的信任！",StringUtils.join(studentNames,"、"), StringUtils.join(classNames,"\",\""));
                        smsService.send(Arrays.asList(customerUser.getPhoneNo()) , content );

                    }catch (Exception e){
                        log.error("消息处理失败,topic=order_success_pay_notify,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER : ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }

                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;

            }
        });

        consumer.subscribe("order_pay_success", "mobileCourseOrder", "OuterOrder", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt : list){
                    try{
                        JSONObject jsonObject = JSONS.toJSONObject(messageExt.getBody());
                        Long payOrderId  = jsonObject.getLong("payOrderId");
                        Long outerOrderId  = jsonObject.getLong("busId");
                        BigDecimal amount = jsonObject.getBigDecimal("amount");
                        JSONObject payInfoExt = new JSONObject();
                        payInfoExt.put("transactionNo" ,jsonObject.getString("transactionNo"));
                        payInfoExt.put("channel" ,jsonObject.getString("channel"));
                        outerOrderService.onlinePaySuccess(payOrderId,outerOrderId,amount,payInfoExt.toJSONString() );
                    }catch (Exception e){
                        log.error("消息处理失败,topic=order_pay_success,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER : ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }

                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;

            }
        });




        consumer.subscribe("attendance", "courseConsume1", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt : list){
                    try{
                        JSONObject data = JSONS.toJSONObject(messageExt.getBody());
                        Long classLectureId  =  data.getLong("classLectureId");
                        List<AttendClassRecord> attendClassRecords = classLectureService.findAttendClassRecordByClassLectureId(classLectureId);
                        for(AttendClassRecord attendClassRecord : attendClassRecords ){
                            try{
                                courseConsumeService.consumeForAttendClassRecord(attendClassRecord);
                            }catch (Exception e){
                                log.error("consume fail",e );
                            }
                        }
                    }catch (Exception e){
                        log.error("消息处理失败,topic=attendance,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER :ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.subscribe("student_refund_success_audit", "InvoiceService", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt:list){
                    JSONObject jsonObject = JSONS.toJSONObject(messageExt.getBody());
                    Long regId = jsonObject.getLong("regId");
                    try{
                        invoiceService.dealStudentRefund(regId);
                    }catch (Exception e){
                        log.error("退费处理发票信息失败regId:"+regId);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.subscribeBroadcast("allin_pay_success",null, new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt : list){
                    try{
                        PosPayRecord posPayRecord = JSONObject.parseObject(messageExt.getBody(), PosPayRecord.class);
                        final String key  = String.valueOf(posPayRecord.getRequestPayId()) ;
                        synchronized (AdminEnrollApiController.deferredResultMultimap){
                            Collection<DeferredResult<String>> deferredResults = AdminEnrollApiController.deferredResultMultimap.get(key);
                            for(DeferredResult<String> deferredResult : deferredResults){
                                AdminEnrollApiController.setDeferredResult(deferredResult ,"Y");
                            }
                        }
                    }catch (Exception e){
                        log.error("消息处理失败,topic=allin_pay_success,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER :ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });


        consumer.subscribe("create_order_success", "create_attendance", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                for(MessageExt messageExt : list){
                    try{
                        Long courseOrderId =  Long.valueOf(new String( messageExt.getBody() ) );

                        List<StudentReg> studentRegList = studentRegService.findStudentRegByCourseOrderId(courseOrderId);

                        if(studentRegList == null || studentRegList.isEmpty() ){
                            log.warn("Can find studentReg : courseOrderId={} " , courseOrderId );
                            if(messageExt.getReconsumeTimes() < 5 ){
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER ;
                            }
                        }else{
                            classLectureService.createAttendanceRecordForStudent(studentRegList);
                            continueClassStatService.calContinueClassRate(studentRegList);
                            List<Long> studentIds  = BeanUtils.toList(studentRegList,"studentId");
                            studentService.updateRegClassAmount(Utils.unique(studentIds));
                        }

                    }catch (Exception e){
                        log.error("消息处理失败,topic=create_order_success,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER :ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.subscribe("create_order_success", "merge_reg_amount", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                for(MessageExt messageExt : list){
                    try{
                        Long courseOrderId =  Long.valueOf(new String( messageExt.getBody() ) );

                        List<StudentReg> studentRegList = studentRegService.findStudentRegByCourseOrderId(courseOrderId);

                        if(studentRegList == null || studentRegList.isEmpty() ){
                            log.warn("Can find studentReg : courseOrderId={} " , courseOrderId );
                            if(messageExt.getReconsumeTimes() < 5 ){
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER ;
                            }
                        }else{
                            for(StudentReg studentReg : studentRegList ){
                                courseClassService.mergeDegree(studentReg.getClassId());
                            }
                        }

                    }catch (Exception e){
                        log.error("消息处理失败,topic=create_order_success,failTime=" + messageExt.getReconsumeTimes() , e );
                        return messageExt.getReconsumeTimes() < 10 ? ConsumeConcurrentlyStatus.RECONSUME_LATER :ConsumeConcurrentlyStatus.CONSUME_SUCCESS ;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.subscribe("student_refund_success_audit", "merge_reg_amount", new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for(MessageExt messageExt:list){
                    JSONObject jsonObject = JSONS.toJSONObject(messageExt.getBody());
                    Long regId = jsonObject.getLong("regId");
                    try{
                        StudentReg studentReg = studentRegService.findById(regId);
                        courseClassService.mergeDegree(studentReg.getClassId());
                    }catch (Exception e){
                        log.error("退费处理处理合并学位regId:"+regId);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

    }

}
